package HandshakeJoin;

import java.util.List;
import java.util.ArrayList;
import java.util.regex.Pattern;
import java.io.IOException;
import java.net.URL;
import com.google.common.base.Charsets;
import com.google.common.io.Resources;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.google.common.primitives.Ints;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Configure
{
    transient private static final Logger logger = LoggerFactory
            .getLogger(Configure.class);

    private int joinId = 0;
    private int numJoinWorkers = 4;
    private int capacityOfWindowR = 256;
    private int capacityOfWindowS = 256;
    private int expiryInterval = 1000;
    private boolean randomSource = false;
    private int ingestInterval = 100;
    private long tupleRLifespan = 1000;
    private long tupleSLifespan = 1000;
    private int snapshotThrh = 256;
    private int snapshotInterval = 2000;
    private int delayCount = 5;
    private boolean blockedTrans = false;
    private int blkSizeR = 16;
    private int blkSizeS = 16;
    private int numFieldsR = 4;
    private int numFieldsS = 4;
    private boolean multimapJoin = false;
    private boolean bstJoin = false;
    private String dbChoice = null;
    private ArrayList<JoinConfig> jConfigs = null;

    public Configure() {}

    /**
     * Parse the join description file.
     * 
     * @param url
     *            Access URL to the join description file.
     */
    public Configure parseJoinDespFile(final URL url) {
        List<String> lines = getLines(url);

        parseEntries(lines);

        return this;
    }

    /**
     * Retrieve joins description lines.
     * 
     * @param url
     *            Access URL to the join description file.
     * @return A list of strings, each of which describes a join.
     */
    public static List<String> getLines(final URL url) {
        List<String> lines = null;
        try {
            lines = Resources.readLines(url, Charsets.UTF_8);
        }
        catch (IOException e) {
            e.printStackTrace();
        }

        return lines;
    }

    public static List<String> removeLinesTillNonempty(List<String> lines) {
        while (!lines.isEmpty()) {
            if (lines.get(0).isEmpty()) {
                lines.remove(0);
            }
            else {
                break;
            }
        }

        return lines;
    }

    public static List<String> removeLinesTillSeparatorBar(List<String> lines,
            final Pattern separatorBar) {
        while (!lines.isEmpty()) {
            if (separatorBar.matcher(lines.remove(0)).matches()) {
                break;
            }
        }

        return lines;
    }

    public static List<String> removeLinesTillSeparatorBar(List<String> lines) {
        Pattern separatorBar = Pattern.compile("\\A\\p{Punct}+\\z");

        return removeLinesTillSeparatorBar(lines, separatorBar);
    }

    /**
     * Parse the join description entries.
     * 
     * @param lines
     *            Lines of the join description file.
     */
    private void parseEntries(List<String> lines) {
        /* Parse the lines for handshake join topology. */
        removeLinesTillSeparatorBar(lines);
        parseEntriesForHandshakeJoinTopology(lines);
        removeLinesTillSeparatorBar(lines);

        /* Parse the lines for join descriptions. */
        removeLinesTillSeparatorBar(lines);
        parseEntriesForJoinConfigs(lines);
    }

    public static String parseOneLine(String line) {
        Iterable<String> split = Splitter.onPattern(":|=").omitEmptyStrings()
                .trimResults().split(line);
        ArrayList<String> terms = Lists.newArrayList(split);

        return terms.get(1);
    }

    private void parseEntriesForHandshakeJoinTopology(List<String> entries) {
        final int numEntriesToParse = 19;

        List<String> args = Lists.newArrayListWithCapacity(numEntriesToParse);
        for (int i = 0; i < numEntriesToParse; ++i) {
            args.add(parseOneLine(entries.get(i)));
        }

        int i = 0;
        numJoinWorkers = Integer.parseInt(args.get(i++));
        capacityOfWindowR = Integer.parseInt(args.get(i++));
        capacityOfWindowS = Integer.parseInt(args.get(i++));
        expiryInterval = Integer.parseInt(args.get(i++));
        randomSource = Boolean.parseBoolean(args.get(i++));
        ingestInterval = Integer.parseInt(args.get(i++));
        tupleRLifespan = Long.parseLong(args.get(i++));
        tupleSLifespan = Long.parseLong(args.get(i++));
        snapshotThrh = Integer.parseInt(args.get(i++));
        snapshotInterval = Integer.parseInt(args.get(i++));
        delayCount = Integer.parseInt(args.get(i++));
        blockedTrans = Boolean.parseBoolean(args.get(i++));
        blkSizeR = Integer.parseInt(args.get(i++));
        blkSizeS = Integer.parseInt(args.get(i++));
        numFieldsR = Integer.parseInt(args.get(i++));
        numFieldsS = Integer.parseInt(args.get(i++));
        multimapJoin = Boolean.parseBoolean(args.get(i++));
        bstJoin = Boolean.parseBoolean(args.get(i++));
        dbChoice = args.get(i++).trim().toLowerCase();
    }

    private void parseEntriesForJoinConfigs(List<String> entries) {
        Pattern separatorBar = Pattern.compile("\\A\\p{Punct}+\\z");

        jConfigs = Lists.newArrayListWithCapacity(entries.size());
        for (int i = 0; i < entries.size(); ++i) {
            String entry = entries.get(i);
            if (entry.isEmpty()) {
                // omit empty lines
                continue;
            }
            if (separatorBar.matcher(entry).matches()) {
                // end of the description
                break;
            }
            if (entry.startsWith("//")) {
                // omit commented lines
                continue;
            }

            Iterable<String> split = Splitter.onPattern(",").omitEmptyStrings()
                    .trimResults().split(entry);
            ArrayList<String> joinArgs = Lists.newArrayList(split);
            logger.info(joinArgs.toString());

            JoinConfig jc = parseJoinArgs(joinArgs);

            jConfigs.add(jc);
        }
    }

    /**
     * Parse join arguments.
     * 
     * @param joinArgs
     *            Join arguments.
     * @return An join object.
     */
    private JoinConfig parseJoinArgs(final ArrayList<String> joinArgs) {
        int i = 0;
        final String name = joinArgs.get(i++);
        final int joinId = generateJoinId();
        final int joinFieldInR = Integer.parseInt(joinArgs.get(i++));
        final int joinFieldInS = Integer.parseInt(joinArgs.get(i++));

        /* Construct a join. */
        JoinConfig jc = new JoinConfig(name, joinId, joinFieldInR, joinFieldInS);

        return jc;
    }

    /**
     * Generate a join ID.
     */
    private int generateJoinId() {
        return ++joinId;
    }

    public ArrayList<JoinConfig> getJoinConfigs() {
        return jConfigs;
    }

    public int getNumJoinWorkers() {
        return numJoinWorkers;
    }

    public int getCapacityOfWindowR() {
        return capacityOfWindowR;
    }

    public int getCapacityOfWindowS() {
        return capacityOfWindowS;
    }

    public int getExpiryInterval() {
        return expiryInterval;
    }

    public boolean getEnableRandomSource() {
        return randomSource;
    }

    public int getIngestInterval() {
        return ingestInterval;
    }

    public long getTupleRLifespan() {
        return tupleRLifespan;
    }

    public long getTupleSLifespan() {
        return tupleRLifespan;
    }

    public int getSnapshotThreshold() {
        return snapshotThrh;
    }

    public int getSnapshotInterval() {
        return snapshotInterval;
    }

    public int getSnapshotDelayCount() {
        return delayCount;
    }

    public boolean getEnableBlockedTransfer() {
        return blockedTrans;
    }

    public int getTupleBlockSizeR() {
        return blkSizeR;
    }

    public int getTupleBlockSizeS() {
        return blkSizeS;
    }

    public int getNumberOfFieldsInTupleR() {
        return numFieldsR;
    }

    public int getNumberOfFieldsInTupleS() {
        return numFieldsS;
    }

    public boolean getEnableMultimapJoin() {
        return multimapJoin;
    }

    public boolean getEnableBSTJoin() {
        return bstJoin;
    }

    public String getDBChoice() {
        return dbChoice;
    }
}